package com.changdu.seatunnel.admin;

import java.time.LocalDate;
import java.util.Collection;
import java.util.HashMap;
import java.util.Map;
import java.util.Objects;
import java.util.Set;

import org.springframework.http.HttpEntity;
import org.springframework.http.HttpHeaders;
import org.springframework.http.MediaType;
import org.springframework.web.client.RestTemplate;

import com.changdu.seatunnel.admin.entity.SyncConfig;
import com.changdu.seatunnel.admin.entity.SyncTask;
import com.changdu.seatunnel.admin.pojo.TableInfo;
import com.changdu.seatunnel.admin.util.JsonUtils;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.node.ArrayNode;

public class JiGuangTaskSync {

	private final RestTemplate restTemplate = new RestTemplate();	
	
	public void convertSyncTask(String sql, Long syncId, String domain) {
        Map<String, Object> requestBody = Map.of(
            "sqlContent", sql,
            "dbType", "mysql",
            "sqlInstanceId", "1854101669915267072",
            "dbName", "bd_actualtime_datalink",
            "limitNum", 10000,
            "originSqlContent", sql
        );
            
        HttpEntity<Map<String, Object>> request = new HttpEntity<>(requestBody, createHeaders());
        JsonNode response = restTemplate.postForObject("https://kunlun-api.changdu.vip/sql-query/query", request, JsonNode.class);
        
        ArrayNode rows = (ArrayNode) response.get("data").get("rows");
        
        Map<String, SyncConfig> result = new HashMap<>();
        for (JsonNode row : rows) {
        	String topic = row.get(0).asText();
        	SyncConfig config = result.get(topic);
        	if (Objects.isNull(config)) {
        		config = new SyncConfig();
        		config.setCreateTime(LocalDate.now().toString());
        		config.setUpdateTime(LocalDate.now().toString());
        		config.setSourceDataCenter("fzidc");
        		config.setTargetDataCenter("fzidc");
        		config.setKafkaTopic(topic);
        		config.setKafkaPartition(3);
        		config.setTableMappings("[]");
        		config.setSourceTemplate(6);
        		config.setTargetTemplate(5);
        		
        		JsonNode sourceInfo = JsonUtils.fromJson(row.get(1).asText(), JsonNode.class);
        		String sourceName = sourceInfo.get("sourceName").asText();
        		if (sourceName.equals("old_tidb_source")) {
        			config.setSourceInstance("idc-tidb-查询");
        		} else if (sourceName.equals("new_tidb_source")) {
        			config.setSourceInstance("idc-adstidb-查询");
        		}
        		config.setTargetInstance("ads-starrocks");
        	}
        	
        	// source databases;
        	Set<String> sourceDatabases = JsonUtils.fromJson(Objects.isNull(config.getSourceDatabases()) ? "[]" : config.getSourceDatabases() , Set.class);
        	Set<TableInfo> sourceTables = JsonUtils.fromJson(Objects.isNull(config.getSourceTables()) ? "[]" : config.getSourceTables(), Set.class);
        	
        	ArrayNode sourceInfos = (ArrayNode) JsonUtils.fromJson(row.get(1).asText(), JsonNode.class).get("dataSourceInfos");
        	for (JsonNode tableInfo : sourceInfos) {
        		String dbName = tableInfo.get("schemaName").asText();
        		String tName = tableInfo.get("tableName").asText();
        		TableInfo t = new TableInfo();
        		t.setRegex(false);
        		t.setDatabase(dbName);
        		t.setName(tName);
        		sourceDatabases.add(dbName);
        		sourceTables.add(t);
			}
        	
        	config.setSourceDatabases(JsonUtils.toJson(sourceDatabases));
        	config.setSourceTables(JsonUtils.toJson(sourceTables));
        	
        	// target databases;
        	String targetDbName = row.get(2).asText();
        	String targetTbName = row.get(3).asText();
        	Set<String> targetDatabases = JsonUtils.fromJson(Objects.isNull(config.getTargetDatabases()) ? "[]" : config.getTargetDatabases(), Set.class);
        	targetDatabases.add(targetDbName);
        	
        	// target tables
        	Set<TableInfo> targetTables = JsonUtils.fromJson(Objects.isNull(config.getTargetTables()) ? "[]" : config.getTargetTables(), Set.class);
        	TableInfo targetTable = new TableInfo();
        	targetTable.setRegex(false);
        	targetTable.setDatabase(targetDbName);
        	targetTable.setName(targetTbName);
        	targetTables.add(targetTable);
        	
        	config.setTargetDatabases(JsonUtils.toJson(targetDatabases));
        	config.setTargetTables(JsonUtils.toJson(targetTables));
        	
        	result.put(topic, config);
        }
		
        insertToDb(result.values(), syncId, domain);
        
	}
	
    private void insertToDb(Collection<SyncConfig> configs, Long syncId, String domain) {
    	HttpHeaders headers = new HttpHeaders();
        headers.setContentType(MediaType.APPLICATION_JSON);
    	
    	for (SyncConfig syncConfig : configs) {
    		syncConfig.setId(syncId);
    		syncConfig.setStatus(1);
    		
			SyncTask syncTask = new SyncTask();
			syncTask.setDomain("https://fuxi-seatunnel-master-none-fzidc.changdu.vip");
			syncTask.setName("kafka=>starrocks:fzidc." + syncConfig.getKafkaTopic());
			syncTask.setShowName(syncTask.getName());
			syncTask.setSourcePlugin("kafka");
			syncTask.setSinkPlugin("starrocks");
			syncTask.setStatus(0);
			syncTask.setSyncConfigId(syncId);
			
			
			JsonNode syncConfigForObject = restTemplate.postForObject(String.format("%s/%s", domain, "api/sync/config"), syncConfig, JsonNode.class);
			System.out.println(syncConfigForObject);
			
			JsonNode syncTaskForObject = restTemplate.postForObject(String.format("%s/%s", domain, "api/task"), syncTask, JsonNode.class);
			System.out.println(syncTaskForObject);
			syncId++;
		}
    	
	}
    
	private HttpHeaders createHeaders() {
        HttpHeaders headers = new HttpHeaders();
        headers.setContentType(MediaType.APPLICATION_JSON);
        headers.set("Accesstoken", "eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9.eyJhdWQiOiIxNTM5NTMiLCJwcmVmZXJyZWRfdXNlcm5hbWUiOiLnjovmmZPkuq4iLCJleHAiOjE3NzMyODUyNDUsImlhdCI6MTc0NzM2NTI0NSwiZW1haWwiOiIxNTM5NTNAbWFpbC5jZCJ9.Z7FPQ5JSYkC3dGTlFH-40YW4wH-nVOu4srMxyMUvXPU");
        return headers;
    }

	public static void main(String[] args) {
		String domain = "https://seatunnel-admin-server-none-prod.changdu.vip";
//		String domain = "http://localhost:9000";
		String sql = "select a.kafka_topic,a.source_info_json,a.target_schema,a.target_table from bd_actualtime_datalink.actual_time_data_link  a  where target_source_name = \"new_starrocks_source\" and online =1 order by kafka_topic;";
		JiGuangTaskSync jiGuangTaskSync = new JiGuangTaskSync();
		jiGuangTaskSync.convertSyncTask(sql, 200L, domain);
	}
	
}
